package TopN;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.Iterator;
import java.util.Set;
import java.util.TreeMap;

//本例子用于选出总流量最大的10个用户及其上下行流量和总流量。

public class topnMapper extends Mapper<LongWritable, Text,Text,FlowBean> {
    TreeMap<FlowBean,Text> treeMap=new TreeMap<>();   //key不能重复，value可以重复，可以重复的话后赋值的那个就会把前赋值的那个给覆盖了    TreeMap只给key排序
    //要进行排序的是Bean对象，看看谁的总流量大

    //
    //每个切片对应一个Mapper类
    //里面每一组key调用一次map方法

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] fields=line.split("\t");
        String phoneNum=fields[0];
        long upFlow=Long.parseLong(fields[1]);
        long downFlow=Long.parseLong(fields[2]);

        Text valueOfMap=new Text(phoneNum);
        FlowBean keyOfMap=new FlowBean(upFlow,downFlow);
        treeMap.put(keyOfMap,valueOfMap);

        //目前是按照降序排序的
        if(treeMap.size()>10){
            treeMap.remove(treeMap.lastKey());
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        //因为treeMap不能通过迭代器或foreach遍历，所以才会有以下的过程
        Set<FlowBean> flowBeans = treeMap.keySet();    //可以返回key的set集合
        //然后通过迭代器依次取出这些key，然后再通过key查询出treeMap里面的value然后输出
        Iterator<FlowBean> iterator = flowBeans.iterator();
        while(iterator.hasNext()){
            FlowBean flowBean = iterator.next();
            Text text=treeMap.get(flowBean);

            context.write(text,flowBean);           //Mapper的输出中key是手机号，value是bean对象
        }
    }
}
